package service

import (
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"iot-bridge/iot_cmd_up/distribute/conf"
	"iot-bridge/iot_cmd_up/distribute/dao/producer"
	"iot-bridge/iot_cmd_up/distribute/dao/redis"
	"iot-bridge/iot_cmd_up/distribute/model"
	"strconv"
	"time"

	log "common/log4go.v1"
)

const (
	TcpReportName = "out|cid|socket|"
	reportTopic   = "iot_cmd_up_"
	reportKey     = "iot_cmd_up"
	CommonUpCmd   = "91"
)

var ErrIllegalPack = errors.New("包不合法")

type kafkaPackBody struct {
	Cmd       int    `json:"cmd"`
	Cid       string `json:"cid"`
	Ts        int64  `json:"ts,string"`
	MessageId int16  `json:"message_id"`
	Payload   []byte `json:"payload"`
}

type TcpReportService struct {
}

func NewTcpReportService() *TcpReportService {
	service := &TcpReportService{}
	return service
}
func (this *TcpReportService) ReportToRedis(sn string, payload string) error {
	timestamp := time.Now().Unix()
	info := &model.RedisReportInfo{
		Sn:        sn,
		TimeStamp: timestamp,
		Payload:   payload,
	}
	iotReportData, err := json.Marshal(info)
	if err != nil {
		log.Error("json.Marshal(%v) error(%v)", info, err)
		return err
	}
	iotReportKey := sn
	iotReportTimestampKey := iotReportKey + "|" + strconv.FormatInt(timestamp, 10)
	err = redis.RedisHSet(TcpReportName, iotReportTimestampKey, string(iotReportData))
	if err != nil {
		log.Error("redis.RedisSet(%s,%s,%s)", TcpReportName, iotReportTimestampKey, string(iotReportData))
		return err
	}
	//存redis list
	err = redis.RedisRPush(iotReportTimestampKey, string(iotReportData))
	if err != nil {
		log.Error("redis.RedisRPush(%s, %s) error(%v)", iotReportTimestampKey, string(iotReportData), err)
		return err
	}
	return nil
}

func (this *TcpReportService) DataToDb(token, table string, reportType int, cid string, sn string, payload []byte) error {
	err := VerifyToken(cid, token)
	if err != nil {
		//todo 验证token 判断错误
		log.Error("cid(%s) token(%s) error(%v)", cid, token, err)
		return err
	}
	timestamp := time.Now().Unix()
	data := hex.EncodeToString(payload)
	info := &model.ReportData{
		ReportType: reportType,
		Cid:        cid,
		Sn:         sn,
		Token:      token,
		TimeStamp:  timestamp,
		Payload:    data,
	}
	_, err = ReportDataDao.SaveReportDataNew(table, info)
	if err != nil {
		log.Error("ReportDataDao.SaveReportDataNew(%s, %v) error(%v)", table, info, err)
		return err
	}
	return nil
}

func (this *TcpReportService) DataToKafka(cid string, payload []byte) error {
	index := 0
	lenPayload := len(payload)
	log.Info("payload(%v) lenPayload(%d)", payload, lenPayload)
	for {
		len, count := packetLen(payload)
		log.Info("len(%d) count(%d)", len, count)
		if int(len+count) > lenPayload {
			log.Error("payload长度有问题")
			return ErrIllegalPack
		}
		argPayload := payload[count : count+len]

		topic, kafkaBody, err := anaPack(cid, argPayload)
		if err != nil {
			log.Error("anaPack(%s, %v) failed error(%v)", cid, argPayload)
			return err
		}

		err = producer.Send(topic, cid, kafkaBody)
		if err != nil {
			log.Error("producer.Send(%s, %s ,%s ) error(%v)", topic, cid, kafkaBody)
			return err
		}
		log.Info("---|||||---producer.Send(%s, %s ,%s ) \n", topic, cid, kafkaBody)
		index += int(count) + int(len)
		if index >= lenPayload {
			break
		} else {
			payload = payload[int(len+count):]
		}
	}
	return nil
}

func anaPack(cid string, argPayload []byte) (topic string, kafkaBody string, err error) {

	//如果是 0x91 通用上报指令  ，则需要将数据扔到 其 产品 对应 的 topic
	if hex.EncodeToString(argPayload[0:1]) == CommonUpCmd {
		//get topic by pid
		var updata *UpData
		updata, err = ana91Pack(cid, argPayload)
		if err != nil {
			log.Error("ana91Pack(%s, %v) error(%v)", cid, argPayload, err)
			return
		}
		kafkaData, _ := json.Marshal(updata)
		kafkaBody = string(kafkaData)
		//get kafka info by topic
		kafkaInfo := getKafkaInfoByPid(updata.ProductId)
		if kafkaInfo == nil {
			log.Error("getKafkaInfoByPid(%s) failed ", updata.ProductId)
			err = errors.New("kafka info is nil")
			return
		}
		topic = kafkaInfo.Topic
	} else {
		instruct := argPayload[0]
		ts := HexDec(hex.EncodeToString(argPayload[1:5]))
		messageId := HexDec(hex.EncodeToString(argPayload[5:7]))

		topic = reportTopic + hex.EncodeToString(argPayload[0:1])
		data := &kafkaPackBody{
			Cmd:       int(instruct),
			Ts:        ts,
			Cid:       cid,
			MessageId: int16(messageId),
			Payload:   argPayload,
		}
		kafkaData, _ := json.Marshal(data)
		kafkaBody = string(kafkaData)
	}

	return
}

func getKafkaInfoByPid(pid string) (result *conf.PidKafka) {

	cacheMap := conf.Conf.PidKafkas
	if cacheMap == nil {
		return
	}
	kafkaInfo, ok := cacheMap[pid]
	if !ok {
		return
	}
	if len(kafkaInfo.Address) == 0 || len(kafkaInfo.Topic) == 0 {
		return
	}
	result = kafkaInfo
	return
}

func ana91Pack(cid string, payload []byte) (upData *UpData, err error) {

	payLoadLen := len(payload)
	if payLoadLen == 0 {
		err = packErr
		log.Error("packErr payLoadLen = 0 cid(%s) payload(%v)", cid, payload)
		return
	}
	index := 0

	//cmd
	cmdLen := 1
	if index+cmdLen > payLoadLen {
		err = packErr
		log.Error("packErr index + cmdLen > payLoadLen  cid(%s) payload(%v)", cid, payload)
		return
	}

	cmdData := payload[index : index+cmdLen]
	if cmdData[0] != uint8(Cmd) {
		err = cmdError
		log.Error("packErr cmdData[0] != uint8(Cmd)  cid(%s) payload(%v)", cid, payload)
		return
	}
	index = index + cmdLen

	var (
		ts      uint32
		msgId   uint16
		bizTime uint64

		pid     string
		module  string
		version string
		bizData []byte
	)

	//ts
	tsLen := 4
	if index+tsLen > payLoadLen {
		err = packErr
		log.Error("packErr index + tsLen > payLoadLen  cid(%s) payload(%v)", cid, payload)
		return
	}
	tsData := payload[index : index+tsLen]
	ts = binary.BigEndian.Uint32(tsData)
	binary.BigEndian.Uint32(tsData)
	index = index + tsLen

	//msgid
	msdIdLen := 2
	if index+msdIdLen > payLoadLen {
		err = packErr
		log.Error("packErr index + msdIdLen > payLoadLen  cid(%s) payload(%v)", cid, payload)
		return
	}
	msgIdData := payload[index : index+msdIdLen]
	msgId = binary.BigEndian.Uint16(msgIdData)
	index = index + msdIdLen

	//pid
	var pidLenData []byte
	for index < payLoadLen {
		item := payload[index]
		pidLenData = append(pidLenData, item)
		index = index + 1
		if item&flag128 == 0 {
			break
		}
	}
	pidLen := decodeChangedLengthData(pidLenData)

	if pidLen > 0 {
		if index+pidLen > payLoadLen {
			err = packErr
			log.Error("packErr index + pidLen > payLoadLen  cid(%s) payload(%v)", cid, payload)
			return
		}
		pidData := payload[index : index+pidLen]
		pid = string(pidData)
		index = index + pidLen
	}

	//module
	var moduleLenData []byte
	for index < payLoadLen {
		item := payload[index]
		moduleLenData = append(moduleLenData, item)
		index = index + 1
		if item&flag128 == 0 {
			break
		}
	}
	moduleLen := decodeChangedLengthData(moduleLenData)

	if moduleLen > 0 {
		if index+moduleLen > payLoadLen {
			err = packErr
			log.Error("packErr index + moduleLen > payLoadLen  cid(%s) payload(%v)", cid, payload)
			return
		}
		moduleData := payload[index : index+moduleLen]
		module = string(moduleData)
		index = index + moduleLen
	}

	//bizTime
	var bizTimeLenData []byte
	for index < payLoadLen {
		item := payload[index]
		bizTimeLenData = append(bizTimeLenData, item)
		index = index + 1
		if item&flag128 == 0 {
			break
		}
	}
	bizTimeLen := decodeChangedLengthData(bizTimeLenData)

	if bizTimeLen > 0 {
		if index+bizTimeLen > payLoadLen {
			err = packErr
			log.Error("packErr index + bizTimeLen > payLoadLen  cid(%s) payload(%v)", cid, payload)
			return
		}
		bizTimeData := payload[index : index+bizTimeLen]

		bizTime, err = bytesToInt(bizTimeData)
		if err != nil {
			return
		}
		index = index + bizTimeLen
	}

	//version
	var versionLenData []byte
	for index < payLoadLen {
		item := payload[index]
		versionLenData = append(versionLenData, item)
		index = index + 1
		if item&flag128 == 0 {
			break
		}
	}
	versionLen := decodeChangedLengthData(versionLenData)

	if versionLen > 0 {
		if index+versionLen > payLoadLen {
			err = packErr
			log.Error("packErr index + versionLen > payLoadLen  cid(%s) payload(%v)", cid, payload)
			return
		}
		versionData := payload[index : index+versionLen]
		version = string(versionData)
		index = index + versionLen
	}

	//bizData
	var bizDataLenData []byte
	for index < payLoadLen {
		item := payload[index]
		bizDataLenData = append(bizDataLenData, item)
		index = index + 1
		if item&flag128 == 0 {
			break
		}
	}
	bizDataLen := decodeChangedLengthData(bizDataLenData)

	if bizDataLen > 0 {
		if index+bizDataLen > payLoadLen {
			err = packErr
			log.Error("packErr index + bizDataLen > payLoadLen  cid(%s) payload(%v)", cid, payload)
			return
		}
		bizDataData := payload[index : index+bizDataLen]
		bizData = bizDataData
		index = index + bizDataLen
	}

	record := &UpData{}
	record.Cid = cid
	record.UpTime = int64(ts)
	record.MessageId = int(msgId)
	record.ProductId = pid
	record.Module = module
	record.ChangeTime = int64(bizTime)
	record.Version = version
	record.Data = bizData

	upData = record
	return

}

type UpData struct {
	Cid       string `json:"cid"`
	UpTime    int64  `json:"up_time"`
	MessageId int    `json:"message_id"`

	ProductId  string `json:"product_id"`         //must  产品id  用于区分接入的硬件  能够使用一套核心数据 和  协议的 硬件,比如手表 watch  ,  ping  可以决定 上行时 数据扔向那个业务段
	Module     string `json:"module"`             //must  区分产品id 下不同的 业务模块  比如 闹钟 课堂  媒资  等
	ChangeTime int64  `json:"change_time,string"` //must  时间戳  单位 秒  数据变化时间,不断增大  ,设备端可能需要根据这个字段判断旧指令
	Version    string `json:"version"`            //maybe 版本号  可以在业务段的一个 模块下 对应唯一的  一份数据 ,操作类指令不需要
	Data       []byte `json:"data"`               //maybe 需要下发的协议数据， 存在时可以在核心数据校验时返回 ，数据不走虚拟设备的模块不需要传参， 比如复杂的通讯录模块
}

var (
	cmdError = errors.New("cmd error")
	packErr  = errors.New("package error")
)

const (
	Cmd     = 0x91
	flag128 = 128 // 1000 0000
	flag127 = 127 // 0111 1111
)

func decodeChangedLengthData(data []byte) (value int) {
	var multiplier = 1
	for _, v := range data {
		digit := int(v)
		value = value + (digit&127)*multiplier
		multiplier = multiplier << 7
		if (digit & 128) != 0 {
			continue
		}
	}
	return
}

func bytesToInt(data []byte) (uint64, error) {

	if len(data) == 3 {
		data = append([]byte{0}, data...)
	} else if len(data) == 5 {
		data = append([]byte{0, 0, 0}, data...)
	} else if len(data) == 6 {
		data = append([]byte{0, 0}, data...)
	} else if len(data) == 7 {
		data = append([]byte{0}, data...)
	}

	switch len(data) {
	case 1:
		return uint64(uint8(data[0])), nil
	case 2:
		return uint64(binary.BigEndian.Uint16(data)), nil
	case 3, 4:
		return uint64(binary.BigEndian.Uint32(data)), nil
	case 5, 6, 7, 8:
		return binary.BigEndian.Uint64(data), nil
	default:
		return uint64(0), fmt.Errorf("%s", "BytesToInt bytes lenth is invaild!")
	}
	return 0, nil
}
